package com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback;

import com.google.common.collect.Lists;
import com.rabbitmq.client.ConfirmListener;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;

import java.util.*;

@Slf4j
public class RabbitMqConfirmListener implements ConfirmListener {

    @Getter
    private SortedMap<Long, Object> confirmMap = Collections.synchronizedSortedMap(new TreeMap<>());

    private List<SuccessCallback> successCallbacks = Lists.newArrayList(new DefaultPublishSuccessCallback());

    private List<PublishFailureCallback> failureCallbacks = Lists.newArrayList(new DefaultFailureCallback());

    @Override
    public void handleAck(long deliveryTag, boolean multiple) {
        if (multiple) {
            Map<Long, Object> map = confirmMap.headMap(deliveryTag + 1);
            map.entrySet().forEach(entry -> invokeCallback(entry.getValue()));
            map.clear();
        } else {
            Object message = confirmMap.remove(deliveryTag);
            invokeCallback(message);
        }
    }

    private void invokeCallback(Object message) {
        if (CollectionUtils.isEmpty(successCallbacks)) {
            return;
        }
        for (SuccessCallback callback : successCallbacks) {
            if (callback == null) {
                continue;
            }
            invokeSuccessCallbacks(message, callback);
        }
    }

    private void invokeSuccessCallbacks(Object rawMessage, SuccessCallback callback) {
        try {
            callback.onSuccess(rawMessage);
        } catch (Exception e) {
            log.error("调用SuccessCallback报错.", e);
        }
    }

    @Override
    public void handleNack(long deliveryTag, boolean multiple) {
        log.warn("Nack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
        if (multiple) {
            confirmMap.headMap(deliveryTag + 1).clear();
        } else {
            confirmMap.remove(deliveryTag);
        }
    }

    public void addSuccessCallback(List<SuccessCallback> successCallbacks) {
        this.successCallbacks.addAll(successCallbacks);
    }

    public void toConfirm(long deliveryTag, Object message) {
        this.confirmMap.put(deliveryTag, message);
    }

    public void remove(long deliveryTag) {
        this.confirmMap.remove(deliveryTag);
    }

    public void addFailureCallback(List<PublishFailureCallback> failureCallbacks) {
        this.failureCallbacks.addAll(failureCallbacks);
    }
}
